package com.edata.bigdata.flink;

import com.edata.bigdata.basic.Commons;
import com.edata.bigdata.flink.annotations.EventTime;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Field;
import java.text.SimpleDateFormat;
import java.util.Date;

public class EventTimeExtractor<T> implements SerializableTimestampAssigner<T> {

    public Logger logger = LoggerFactory.getLogger(this.getClass());
    private final Class<T> clazz;
    // "yyyy-MM-dd HH:mm:ss:SSS";
    private final String dateFormat;

    @Override
    public long extractTimestamp(T evt, long l) {
        try {
            Field f = Commons.findFieldFromBeanByAnnotation(clazz, EventTime.class);
            f.setAccessible(true);
            Object value = f.get(evt);
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS");
            Date date = sdf.parse(value.toString());
            return date.getTime();
        } catch (Exception e) {
            long currentTime = System.currentTimeMillis();
            logger.warn("无法找到 {} 的合法时间字段 {}：使用系统时间：{}", clazz, e.getMessage(), currentTime);
            return currentTime;
        }
    }

    public EventTimeExtractor(Class<T> clazz, String dateFormat) {
        this.clazz = clazz;
        this.dateFormat = dateFormat;
    }
}
